package main

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"os"
	"time"
)

type OrderListener struct {
}

func NewOrderListener() *OrderListener {
	return &OrderListener{}
}

func (dl *OrderListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
	// 下单业务
	// 扣库存 --
	fmt.Printf("开始执行本地逻辑\n")
	time.Sleep(3 * time.Second)
	fmt.Printf("执行本地逻辑成功\n")
	//return primitive.CommitMessageState // 执行成功，可以事务消息，不进入：CheckLocalTransaction
	//return primitive.RollbackMessageState // 执行失败，不发送事务消息，不进入：CheckLocalTransaction
	return primitive.UnknowState // 扣减库存 开始进行回查，不发送事务消息，进入：CheckLocalTransaction
}

func (dl *OrderListener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
	// 订单回查
	fmt.Printf("rocketmq开始执行消息回查")
	time.Sleep(5 * time.Second)
	return primitive.CommitMessageState
}

func main() {
	// 1: 创建一个事务型生产者
	p, _ := rocketmq.NewTransactionProducer(
		NewOrderListener(),
		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"121.199.66.66:9876"})),
		producer.WithRetry(2),
	)

	err := p.Start()
	if err != nil {
		fmt.Println("启动生产者失败,%s", err.Error())
		os.Exit(1)
	}

	// 确定一个主题
	topic := "order_transcation_topic"
	message := "下单了。订单编号是：10000"
	msg := &primitive.Message{
		Topic: topic,
		Body:  []byte(message),
	}

	// 这里选择严格的发送，发送成功就有回执
	res, err := p.SendMessageInTransaction(context.Background(), msg)
	if err != nil {
		fmt.Printf("send message error: %s\n", err)
	} else {
		// 消息发送回执
		fmt.Printf("send message success: result=%s : message=%s\n", res.String(), message)
	}

	time.Sleep(5 * time.Minute)
	err = p.Shutdown()
	if err != nil {
		fmt.Printf("shutdown producer error: %s", err.Error())
	}
}
